package io.mqttpush.mqttserver.middle;

import io.mqttpush.mqttserver.handles.DispatchHandle;
import io.mqttpush.mqttserver.handles.NoticeHandle;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * @author tianzhenjiu
 * @date 2020/4/27 16:35
 */
public class TcpServer {


    Logger logger= LoggerFactory.getLogger(getClass());

    private int sobacklog = 4;

    private int port = 10000;

    private int bossthreads = Runtime.getRuntime().availableProcessors();

    private int workthreads = Runtime.getRuntime().availableProcessors() * 2;

    EventLoopGroup group = null;
    EventLoopGroup workGroup = null;
    ChannelFuture serverFuture;

    public  void start(int port) {

        try{
            start("0.0.0.0",port);
        }catch (Exception e){
            e.printStackTrace();
            stop();
        }

    }


    public void start(String hostName,int port)throws InterruptedException{

        ServerBootstrap bootstrap = new ServerBootstrap();


        if (Epoll.isAvailable()) {
            group = new EpollEventLoopGroup(bossthreads);
            workGroup = new EpollEventLoopGroup(workthreads);
            bootstrap.channel(EpollServerSocketChannel.class);
        } else {
            group = new NioEventLoopGroup(bossthreads);
            workGroup = new NioEventLoopGroup(workthreads);
            bootstrap.channel(NioServerSocketChannel.class);
        }



        bootstrap.group(group, workGroup);
        bootstrap.childOption(ChannelOption.WRITE_SPIN_COUNT, 4);

        // 绑定端口
        bootstrap.localAddress(hostName,port);


        /**
         * 设置sobacklog
         * https://www.jianshu.com/p/e6f2036621f4
         * sobacklog 是指定当前sercersocket的accept队列长度
         */
        bootstrap.option(ChannelOption.SO_BACKLOG, sobacklog);

        bootstrap.childHandler(new ChannelInitializer<Channel>() {
            @Override
            protected void initChannel(Channel ch) throws Exception {
                ch.pipeline()
                        //.addLast(new LoggingHandler(LogLevel.DEBUG))
                        .addLast(MqttEncoder.INSTANCE)
                        .addLast(new MqttDecoder())
                        .addLast(new DispatchHandle())
                        .addLast(new NoticeHandle());


            }
        });


        serverFuture = bootstrap.bind();

        serverFuture.addListener((future)->{

            if(future.isSuccess())
            {
                logger.info("bind the host->{} port->{}  success",hostName,port);
            }

        });
        serverFuture.channel().closeFuture().sync();
    }


    public void stop(){

        if(serverFuture!=null){
            serverFuture.channel().close();
        }

        if(group!=null){
            group.shutdownGracefully(0,1, TimeUnit.MILLISECONDS);
        }

        if(workGroup!=null){
            workGroup.shutdownGracefully(0,1, TimeUnit.MILLISECONDS);
        }

    }
}
